package org.yonggan.dmp.graphx

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}
import org.yonggan.dmp.conf.ConfigManager
import org.yonggan.dmp.tag.EquipmentTag
import org.yonggan.dmp.utils.UserTagUtils

/**
  * 使用图计算 计算用户的画像
  */
object UserGraphTag {

  def main(args: Array[String]): Unit = {

    val config = new SparkConf()
      .setMaster("local[*]")
      .setAppName("处理用户标签")
    val sc = new SparkContext(config)
    val sqlContext = new SQLContext(sc)

    val appdict = sc.textFile(ConfigManager.APP_DICT_FILE_PATH)
      .map(_.split("\t", -1))
      .filter(_.length >= 5)
      .map(arr => (arr(4), arr(1)))
      .collectAsMap()

    // 将数据广播出去
    val appbct = sc.broadcast(appdict)

    /**
      * 处理用户搜索停用词
      */
    val stopWord = sc.textFile(ConfigManager.USER_STOP_WORD)
    val stopWordArr = stopWord.collect()
    val stopWordBct: Broadcast[Array[String]] = sc.broadcast(stopWordArr)

    // 读取parqut 文件
    val pqDF = sqlContext.read.parquet(ConfigManager.PARQUET_OUT)

    // 提取用户数据
    val baseRDD = pqDF.rdd.mapPartitions(itor => {
      val resultItor = itor.map(row => {

        val uid = UserIdentity.allUserId(row)
        // 设备标签
        val equip = EquipmentTag.makeTag(row)
        //关键字
        val keywordList = UserTagUtils.keyWordTag(row, stopWordBct)
        val tagList = equip ++ keywordList
        (uid, tagList)
      })
      resultItor
    }).filter(_._1.size > 0).cache()

    // 构建顶点
    val vertexRDD: RDD[(VertexId, List[(String, Int)])] = baseRDD.mapPartitions(partition => {

      partition.flatMap { tp =>
        // 用户标识
        val allUser = tp._1
        // 提取标签
        var allTag = tp._2

        val result: List[(VertexId, List[(String, Int)])] = allUser.map(user => {
          val tp: (VertexId, List[(String, Int)]) = if (user.equals(allUser.head)) {
            allTag :+= (user, 0)
            (user.hashCode.toLong, allTag)
          } else {
            (user.hashCode.toLong, Nil)
          }
          tp
        })
        result
      }
    })

    // 构建图计算边
    val edgeRDD: RDD[Edge[Int]] = baseRDD.flatMap(tp => {
      // 用户标签
      val userLst = tp._1
    //  userLst.map(f => Edge(userLst.head.hashCode.toLong, f.hashCode.toLong, 0))
      userLst.map(f => Edge(f.hashCode.toLong, userLst.head.hashCode.toLong, 0))
    })

    // 构建图对象
    val graph = Graph(vertexRDD, edgeRDD)

    // 使用graph connectedComponents 算法
    val cc = graph.connectedComponents().vertices

    val resultTag: RDD[(VertexId, List[(String, Int)])] = cc/* map (_.swap)*/ join (vertexRDD) map {
      case (id, (masterId, lst)) => (masterId,lst)
    } reduceByKey {
      (lst1,lst2) => {
        (lst1 ++ lst2) groupBy(_._1) mapValues( _ map(_._2) sum) toList
      }
    }

    resultTag.foreach(println)

    sc.stop()
  }

}
